package com.gzhryc.common.kafka.procuder;


import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.gzhryc.common.kafka.JFKafkaProducer;
import com.gzhryc.common.kafka.models.KafkaMessage;

public class MultipleKafkaProducer implements JFKafkaProducer {

    @SuppressWarnings("unused")
	private Properties properties;
    private BlockingQueue<KafkaMessage> queue;
    
    private List<ProducerWorkThread> producers;
    private AtomicBoolean isRunning;
    
    public MultipleKafkaProducer(Properties properties,Integer threadNum) {
    	this.properties = properties;
        this.queue = new LinkedBlockingQueue<KafkaMessage>();
        this.producers = new ArrayList<ProducerWorkThread>();
        this.isRunning = new AtomicBoolean(true);
        if(threadNum != null && threadNum > 0) {
        	for(int i=0;i<threadNum;i++) {
        		ProducerWorkThread workThread = new ProducerWorkThread(i,properties);
        		workThread.start();
        		producers.add(workThread);
        	}
        }
    }
    
    @Override
	public boolean send(KafkaMessage msg) {
    	if(queue != null && isRunning.get()) {
    		return queue.add(msg);
    	}
		return false;
	}
    
    private class ProducerWorkThread extends Thread {
    	
    	Integer index;
    	Properties properties;
    	KafkaProducer<String, String> producer;
    	
    	public ProducerWorkThread(Integer index,Properties properties) {
    		this.index = index;
    		this.properties = properties;
    	}
    	
    	public void run() {
    		producer = new KafkaProducer<String, String>(properties);
    		while(isRunning.get()) {
    			try {
    				KafkaMessage msg = queue.poll(10, TimeUnit.SECONDS);
    				if(msg != null) {
    					ProducerRecord<String, String> producerRecord = new ProducerRecord<>(msg.getTopic(), msg.getKey(), msg.getValue()) ;
    					producer.send(producerRecord);
    					producer.flush();
    				}
				} catch (Exception e) {
					e.printStackTrace();
				}
    		}
    		if(producer != null) {
    			producer.close();
    		}
    	}
    	
    	@SuppressWarnings("unused")
		public Integer getIndex() {
    		return index;
    	}
    }

	@Override
	public void close() {
		this.isRunning.set(false);
	}

}
